package com.qf.ps;

import com.qf.config.RabbitMQConfig;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 主题模式  --  主题交换机  -- RoutingKey  #  *
 * @author lixu
 */
@Component
public class TopicQueue {

    /**
     * 消费者1  消费队列1   Q1
     */
    @RabbitListener(queues = "Q1")
    public void receiverMessageDirectQueue1(String msg, Channel channel, Message message){//自动确认已经关闭了  改成手动确认了
        try {
            // 方法形参 ;  可以直接接收任意类型的消息
            System.out.println("=====Q1：消费者1：" + msg);
            //抛异常
            int i = 1/0;
            //执行很多的代码
            //手动确认  basic:基础 Ack：应答
            //参数1： long deliveryTag, delivery 投递   Tag：标记
            //参数2： boolean multiple ： 是否批量执行（投递的标记）  不批量应答  只确认当前此次这个消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        } catch (Exception e) {

            //判断 消息是否已经给过机会了    消息是否已经重新投递过了   第二次投递过了
            if(message.getMessageProperties().isRedelivered()){
                //不能再投递  要不然没完没了
                System.out.println("==此消息已经重新投递过，不能再投递了,要不然没完没了 =====");
                //拒绝再投递消息
                 // Reject : 拒绝
                //参数1： long deliveryTag, delivery 投递   Tag：标记
                //参数2： boolean requeue ： 表示是否将消息放回队列  (让我再处理一次）
                // true: 将消息放回队列 让我再处理一次   false: 不放回队列 我不处理了
                try {
                    channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
                    //打印日志 转人工介入  （公司是转人工 调用接口）
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }else{
                //不好意思出问题了  RabbitMQ服务器 你再相信我一次 让我再处理一次
                //手动确认  basic:基础 Not Ack：不应答
                //参数1： long deliveryTag, delivery 投递   Tag：标记
                //参数2： boolean multiple ： 是否批量执行（投递的标记）  不批量应答  只确认当前此次这个消息
                //参数3： boolean requeue ： 表示是否将消息放回队列  (让我再处理一次）
                // true: 将消息放回队列 让我再处理一次   false: 不放回队列 我不处理了
                try {
                    System.out.println("==此消息没有重新投递过，再投递一次 =====");
                    channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
                } catch (IOException ex) {
                    ex.printStackTrace();
                }
            }



        }
        /**
         * Message :
         *   第一部分： 消息的属性  （消息头）
         *   第二部分： 消息的内容 （ 消息体）
         *   -------
         * 请求：
         *   第一部分： 请求头
         *   第二部分： 请求体
         */

    }
    /**
     * 消费者2  消费队列2   Q2
     */
    @RabbitListener(queues = "Q2")
    public void receiverMessageDirectQueue2(String msg){
        // 方法形参 ;  可以直接接收任意类型的消息
        System.out.println("=====Q2：消费者2：" + msg);
    }

    /**
     * 消费者1  消费队列1   boot_topic_q1
     */
    @RabbitListener(queues = RabbitMQConfig.TOPICQ1)
    public void receiverMessageDirectQueue3(String msg){
        // 方法形参 ;  可以直接接收任意类型的消息
        System.out.println("=====boot_topic_q1：消费者1：" + msg);
    }




    @Autowired
    private RedisTemplate redisTemplate;//使用到Redis的Java客户端
    /**
     * 消费者2  消费队列2   Q2 boot_topic_q2
     *
     * 接收到消息
     *  测试避免消息的重复性消息
     *
     *
     *
     */
    @RabbitListener(queues = RabbitMQConfig.TOPICQ2)
    public void receiverMessageDirectQueue4(String msg, Channel channel, Message message) throws Exception{
        // 方法形参 ;  可以直接接收任意类型的消息
        System.out.println("=====boot_topic_q2：消费者2：" + msg);

        // 使用Redis来实现  避免消息的重复性消息  幂等性问题
        String uuidKey = (String) message.getMessageProperties().getHeaders()
                .get("spring_returned_message_correlation");
        //判断redis是否已经存在了  uuid
        if(redisTemplate.opsForValue().setIfAbsent(uuidKey,"0",10, TimeUnit.SECONDS)){
            //设置uuid 同时  表示自己正在执行消息   为了防止死锁、设置了过期时间 10s
            System.out.println("======正在干活中......==============");
            //此消息我已经消费完成   永久存在
            redisTemplate.opsForValue().set(uuidKey,"1");
            //手动应答
            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }else{
            //表示此uuidKey这个消息正在被别的消费者所消费

            //判断此uuidKey这个消息 是否已经被成功的消费了
            if("1".equalsIgnoreCase(String.valueOf(redisTemplate.opsForValue().get(uuidKey)))){
                //已经被完全消费掉了
                //手动应答
                channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
            }else{
                //正在被别的消费者所消费  是否能成功 不知道 只能放弃 并不删除RabbitMQ队列中的消费
                //我们消费不了 我们也不应答
            }
        }


    }
}
